# -*- coding: utf-8 -*-
# standard
import os
import sys
import time
import json
from datetime import datetime
from datetime import timedelta
from typing import List

# third
import requests
from sqlalchemy import case, extract

# local
_P_PATH =  os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
if _P_PATH not in sys.path:
    sys.path.append(_P_PATH)
from models import *
from functions.ruiju_func import *
from functions.tianqing_func import *


"""
@Title:   
@File: influx_func.py
@Author: walle 2024年09日28日 08时44分06秒
@Version: 1.0.0
@Desc: 
* 维恩电子  wn1010.com  数据格式
{
    "uuid":"36FFD4055256343905660743",
    "grp":1,
    "wql":26,
    "nodes":[
        {"id":"1","t":20.6,"bat":6},
        {"id":"2","t":20.5,"bat":6}
        ]
}
36FFD4055256343905660743:WT/ET 的唯一编码，可以用于区分网关设备
Grp:组号
Id: 组内编号
Wql:WIFI 版本的信号强度
t:温度
H:湿度
bat:电量
数据类型详细
"""


def list_temperature_iot_hosts(bucket_name: str = None, table_name: str = None) -> List[dict]:
    """
    列出沈阳维恩的温度传感器主机设备的uuid
    * 去重: distinct(column: "_value")
    * 保留指定的列: keep(columns: ["_time", "field1", "field2"])
    """
    bucket_name = "wn1010" if bucket_name is None else bucket_name
    table_name = "temperature" if table_name is None else table_name
    query_api = INFLUX_CLI.query_api()
    query = f"""from(bucket:"{bucket_name}")
      |> range(start: 0)
      |> filter(fn: (r) => r._measurement == "{table_name}" and r._field != "token")
      |> keep(columns: ["uuid", "_measurement"])
      |> distinct(column: "uuid")
      """
    tables: List[flux_table.FluxTable] = query_api.query(query, org=INFLUXDB_ORG)
    data = []
    for table in tables:
        for record in table.records:
            current_table_name = record.get_measurement()
            uuid = record.values.get("uuid")
            data.append({"table_name": current_table_name, "uuid": uuid})
    return data


def get_iot_temperature_data(bucket_name: str| None = None, table_name: str| None = None, 
                             uuids: List[str] | None = None, group_ids: List[str] | None = None, 
                             sensor_ids: List[str] | None = None, to_local_time: int = 0,
                             start: datetime| None = None, max_point: int = 1000,
                             stop: datetime| None = None) -> Resp:
    """
    获取温度传感器温度数据
    :param bucket_name: 数据库
    :param table_name: 表名
    :param uuids: uuid列表
    :param start: 开始时间
    :param stop: 结束时间
    :param max_point: 最多分多少个数据， 会影响返回速度和精度，也会决定返回数据的最大数量
    :return:
    """
    bucket_name = "wn1010" if bucket_name is None else bucket_name
    table_name = "temperature" if table_name is None else table_name
    start = datetime.now() - timedelta(days=1) if start is None else start
    stop = datetime.now() if stop is None else stop
    interval = int((stop - start).total_seconds() / max_point / 60)
    logger.debug(f"设置聚合时间间隔为 {interval} 分钟")
    start = start.strftime("%Y-%m-%dT%H:%M:%SZ")
    stop = stop.strftime("%Y-%m-%dT%H:%M:%SZ")
    query_api = INFLUX_CLI.query_api()
    query = f"""from(bucket:"{bucket_name}")
      |> range(start: {start}, stop: {stop})
      |> filter(fn: (r) => r._measurement == "{table_name}")
      """
    if uuids:
        'r["uuid"] == "30FFD2053054483209770343" or r["uuid"] == "30FFCF053054483213600343"'
        container = [f'r["uuid"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    if group_ids:
        container = [f'r["grp"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    if sensor_ids:
        container = [f'r["id"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    query += f'|> filter(fn: (r) => r["_field"] == "t")\n'  # 只取温度数据
    query += f'|> keep(columns: ["_time", "_value", "uuid", "grp", "id"])\n'  # 只保留时间、温度、uuid、grp、id字段 必须放在 filter之后
    query += f'|> aggregateWindow(every: {interval}m, fn: mean, createEmpty: false)\n'  # 每隔2分钟取一次平均值
    query += f'|> yield(name: "mean")\n'  # 将结果命名为mean
    tables: List[flux_table.FluxTable] = query_api.query(query, org=INFLUXDB_ORG)
    if to_local_time:
        data = [{
            "uuid": record["uuid"],
            "group": record["grp"],
            "sensor_id": record["id"],
            "time": convert_to_local_time(record["_time"]),
            "value": record["_value"],
        } for table in tables for record in table.records]
    else:
        data = [{
            "uuid": record["uuid"],
            "group": record["grp"],
            "sensor_id": record["id"],
            "time": record["_time"].strftime("%Y-%m-%d %H:%M:%S"),
            "value": record["_value"],
        } for table in tables for record in table.records]
    return data


def get_iot_battery_data(bucket_name: str| None = None, table_name: str| None = None, 
                             uuids: List[str] | None = None, group_ids: List[str] | None = None, 
                             sensor_ids: List[str] | None = None, to_local_time: int = 0,
                             start: datetime| None = None,
                             stop: datetime| None = None) -> Resp:
    """
    获取温度传感器电池数据 6是满电
    :param bucket_name: 数据库
    :param table_name: 表名
    :param uuids: uuid列表
    :param start: 开始时间
    :param stop: 结束时间
    :return:
    """
    bucket_name = "wn1010" if bucket_name is None else bucket_name
    table_name = "temperature" if table_name is None else table_name
    start = datetime.now() - timedelta(days=1) if start is None else start
    stop = datetime.now() if stop is None else stop
    start = start.strftime("%Y-%m-%dT%H:%M:%SZ")
    stop = stop.strftime("%Y-%m-%dT%H:%M:%SZ")
    query_api = INFLUX_CLI.query_api()
    query = f"""from(bucket:"{bucket_name}")
      |> range(start: {start}, stop: {stop})
      |> filter(fn: (r) => r._measurement == "{table_name}")
      """
    if uuids:
        'r["uuid"] == "30FFD2053054483209770343" or r["uuid"] == "30FFCF053054483213600343"'
        container = [f'r["uuid"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    if group_ids:
        container = [f'r["grp"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    if sensor_ids:
        container = [f'r["id"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    query += f'|> filter(fn: (r) => r["_field"] == "bat")\n'  # 只取电池温度数据
    query += f'|> keep(columns: ["_time", "_value", "uuid", "grp", "id"])\n'  # 只保留时间、温度、uuid、grp、id字段 必须放在 filter之后
    query += '|> last()'  # 取最后一个数据
    tables: List[flux_table.FluxTable] = query_api.query(query, org=INFLUXDB_ORG)
    if to_local_time:
        data = [{
            "uuid": record["uuid"],
            "group": record["grp"],
            "sensor_id": record["id"],
            "time": convert_to_local_time(record["_time"]),
            "value": record["_value"],
        } for table in tables for record in table.records]
    else:
        data = [{
            "uuid": record["uuid"],
            "group": record["grp"],
            "sensor_id": record["id"],
            "time": record["_time"].strftime("%Y-%m-%d %H:%M:%S"),
            "value": record["_value"],
        } for table in tables for record in table.records]
    return data
    
    
DEFAULT_CUSTOMER_ID = 76  # 麦上包子
DEFAULT_STORE_ID = 30  # 控江路店


async def save_sensor_data(data_type: str,
                           uuid: str,
                           grp: int,
                           nodes: list,
                           customer_id: int,
                           store_id: int):
    """
    保存数据到influxdb
    :param data_type: 车辆的数据类型, 默认 temperature 表示温度
    :param uuid:  网关唯一编码
    :param grp: 组号
    :param wql: wifi信号强度
    :param nodes: 传感器节点数据
    """
    t1 = time.perf_counter()
    bucket = "wn1010" if customer_id == DEFAULT_CUSTOMER_ID else generate_customer_bucket_name(customer_id)
    points = []
    for node in nodes:
        point = Point(data_type)
        point.tag("customer_id", customer_id)
        point.tag("store_id", store_id)
        point.tag("uuid", uuid)
        point.tag("grp", grp)
        point.tag("id", node["id"])
        point.field("t", node["t"])
        point.field("bat", node["bat"])
        points.append(point)
    t2 = time.perf_counter()
    influx_db_client_async: InfluxDBClientAsync = await get_influxdb_cli_async()
    write_api: WriteApiAsync = influx_db_client_async.write_api()
    t3 = time.perf_counter()
    try:
        await write_api.write(bucket=bucket, org=INFLUXDB_ORG, record=points)
    except Exception as e:
        logger.error(e)
    finally:
        t4 = time.perf_counter()
        InFluxDBAlertOrm.recovery_offline_alert(uuid=uuid, group_id=grp, nodes=nodes)
        t5 = time.perf_counter()
        ms = f"获取influx client 耗时:{(t2 - t1) * 1000:.2f} ms,获取write_api耗时:{(t3 - t2) * 1000:.2f} ms, 写入influx_db耗时:{(t4 - t3) * 1000:.2f} ms,检查阈值和离线告警耗时:{(t5 - t4) * 1000:.2f} ms, 总耗时:{(t5 - t1) * 1000:.2f} ms"
        logger.debug(ms)
        logger.info(f"t5- t4 = {(t5 - t4)}")
        logger.info(f"t4 - t3 = {(t4 - t3)}")
        logger.info(f"t3 - t2 = {(t3 - t2)}")
        logger.info(f"获取 = {(t2 - t1)}")


def delete_data(bucket: str, org: str, table_name: str):
    # 删除数据
    delete_api = INFLUX_CLI.delete_api()
    start = "1970-01-01T00:00:00Z"
    stop = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
    delete_api.delete(start, stop, f'_measurement="{table_name}"', bucket=bucket, org=org)


def sync_base_demo(table_name: str, bucket: str = None, data_list: List[dict] = None, clear_history: bool = False):
    # 同步写的基础例子
    bucket = "lan_event" if bucket is None else bucket
    if clear_history:
        delete_data(bucket, INFLUXDB_ORG, table_name)
    write_api = INFLUX_CLI.write_api(write_options=SYNCHRONOUS)
    data_list = data_list if data_list is not None else [{"tagname1": "tagvalue1", "field1": 1}, {"tagname1": "tagvalue2", "field1": 2}]
    points = []
    for data in data_list:
        point = Point(table_name)
        for key, value in data.items():
            if isinstance(value, str):
                point.tag(key, value)
            else:
                point.field(key, value)
            point.time(datetime.now())
        point._time = datetime.now()
        points.append(point)
    points = Point(table_name).tag("tagname1", "tagvalue121").field("field1", 100)
    write_api.write(bucket=bucket, org=INFLUXDB_ORG, record=points)
    time.sleep(1) # separate points by 1 second

    query_api = INFLUX_CLI.query_api()

    query = f"""from(bucket: "{bucket}")
    |> range(start: -10m)
    |> filter(fn: (r) => r._measurement == "{table_name}")"""
    tables = query_api.query(query, org=INFLUXDB_ORG)
    print("---------------------------------")
    for table in tables:
        for record in table.records:
            if record.get_measurement() == table_name:
                print(f"record: {record}")
            print(f"record: {record}")

def query_latest_data(bucket: str, measurement: str, minutes: int = 10):
    query_api = INFLUX_CLI.query_api()

    query = f"""from(bucket: "{bucket}")
    |> range(start: -{minutes}m)
    |> filter(fn: (r) => r._measurement == "{measurement}")"""
    tables = query_api.query(query, org="yhjk")

    for table in tables:
        for record in table.records:
            print(f"record: {record}")


def get_sensor_alert_customer_summary_group_by_store_old(
        sensor_category: str,
        customer_id: int | None = None,
    ) -> Resp:
    """
    todo: 准备废止 2024-11-26
    查询温度传感器的警告统计信息，以门店分组，key为品牌，这个函数将来可能会修改或者扩展
    """
    db_session, resp = new_db_session(), Resp()
    event_type = SensorEventType.OFFLINE.value  # 所有类型的传感器都有离线警告而且字面值是一致的
    sub1 = select(
        InFluxDBAlertOrm.store_id,
        func.count(InFluxDBAlertOrm.id).label("alert_history_total"),
        func.sum(case((and_(InFluxDBAlertOrm.status == 1, InFluxDBAlertOrm.type == event_type), 1), else_=0)).label("offline_sensor_count"),
        func.sum(case((and_(InFluxDBAlertOrm.status == 1, InFluxDBAlertOrm.type != event_type), 1), else_=0)).label("threshold_sensor_count"),
        ).where(
            InFluxDBAlertOrm.sensor_category == sensor_category,
            ).group_by(
                InFluxDBAlertOrm.store_id,
            ).subquery()
    if customer_id:
        s = select(
            SensorOrm.customer_id,
            SensorOrm.store_id,
            func.count(SensorOrm.id).label("sensor_count"),
            sub1.c.alert_history_total,
            sub1.c.offline_sensor_count,
            sub1.c.threshold_sensor_count,
        ).outerjoin_from(
            SensorOrm, sub1, SensorOrm.store_id == sub1.c.store_id,
        ).where(
            SensorOrm.customer_id == customer_id,
            SensorOrm.sensor_category == sensor_category,
            ).group_by(
                SensorOrm.customer_id, 
                SensorOrm.store_id,
                sub1.c.alert_history_total,
                sub1.c.offline_sensor_count,
                sub1.c.threshold_sensor_count,
                )
    else:
        s = select(
            SensorOrm.customer_id,
            SensorOrm.store_id,
            func.count(SensorOrm.id).label("sensor_count"),
            sub1.c.alert_history_total,
            sub1.c.offline_sensor_count,
            sub1.c.threshold_sensor_count,
        ).join_from(
            SensorOrm, sub1, SensorOrm.store_id == sub1.c.store_id,
        ).group_by(
                SensorOrm.customer_id, 
                SensorOrm.store_id,
                sub1.c.alert_history_total,
                sub1.c.offline_sensor_count,
                sub1.c.threshold_sensor_count,
                )
    result = db_session.execute(s).all()
    db_session.close()
    summary = {}
    store_ids = set()
    for customer_id, store_id,  sensor_count, alert_history_total, offline_sensor_count, threshold_sensor_count in result:
        store_container: list = summary.get(customer_id, {})
        store_ids.add(store_id)
        store_container[store_id] = {
            "sensor_count": sensor_count,  # 传感器总数
            "alert_history_total": alert_history_total if alert_history_total else 0,  # 报警总数，包括已恢复的和没恢复的
            "offline_sensor_count": offline_sensor_count if offline_sensor_count else 0,  # 处于离线状态的传感器汇总
            "threshold_sensor_count": threshold_sensor_count if threshold_sensor_count else 0,  # 处于阈值警告状态的传感器汇总
            }
        summary[customer_id] = store_container
    resp.data = {"summary": summary, "store_ids": list(store_ids)}
    return resp


def get_sensor_customer_count(customer_id: int) -> Resp:
    """
    查询客户下所有传感器数量
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
        func.count(SensorOrm.id),
        ).where(
        SensorOrm.customer_id == customer_id,
        SensorOrm.is_deleted == 0,
        )
    result = db_session.execute(s).scalar()
    db_session.close()
    resp.data = result
    return resp


def get_sensor_alert_customer_summary_group_by_store(
        sensor_category: str,
        customer_id: int | None = None,
    ) -> Resp:
    """
    查询温度传感器的警告统计信息，以门店分组，key为品牌，这个函数将来可能会修改或者扩展
    """
    db_session, resp = new_db_session(), Resp()
    sub = select(
        InFluxDBAlertOrm.store_id.label("store_id"),
        func.count(InFluxDBAlertOrm.id).label("alert_history_total"),
    ).where(
        InFluxDBAlertOrm.sensor_category == sensor_category,
    ).group_by(InFluxDBAlertOrm.store_id).subquery()
    s = select(
        SensorOrm.customer_id,
        SensorOrm.store_id,
        func.count(SensorOrm.id).label("sensor_count"),
        sub.c.alert_history_total,
        func.sum(SensorOrm.is_offline).label("offline_sensor_count"),
        func.sum(SensorOrm.is_over_threshold).label("threshold_sensor_count"),
        ).outerjoin_from(
            SensorOrm, sub, sub.c.store_id == SensorOrm.store_id,
        ).where(
            SensorOrm.sensor_category == sensor_category,
            SensorOrm.is_deleted == 0,
            )
    if customer_id:
        s = s.where(SensorOrm.customer_id == customer_id)
    s = s.group_by(SensorOrm.customer_id, SensorOrm.store_id, sub.c.alert_history_total)
    result = db_session.execute(s).all()  
    db_session.close()
    summary = {}
    store_ids = set()
    for customer_id, store_id,  sensor_count, alert_history_total, offline_sensor_count, threshold_sensor_count in result:
        store_container: list = summary.get(customer_id, {})
        store_ids.add(store_id)
        store_container[store_id] = {
            "sensor_count": sensor_count,  # 传感器总数
            "alert_history_total": alert_history_total if alert_history_total else 0,  # 报警总数，包括已恢复的和没恢复的
            "offline_sensor_count": offline_sensor_count if offline_sensor_count else 0,  # 处于离线状态的传感器汇总
            "threshold_sensor_count": threshold_sensor_count if threshold_sensor_count else 0,  # 处于阈值警告状态的传感器汇总
            }
        summary[customer_id] = store_container
    resp.data = {"summary": summary, "store_ids": list(store_ids)}
    return resp


def page_influxdb_alerts(customer_ids: List[int] | None = None,
                         store_ids: List[int] | None = None,
                         uuid: str | None = None,
                         group_id: int | None = None,
                         sensor_id: str | None = None,
                         type: str | None = None,
                         status: int | None = None,
                         priority: str | None = None,
                         sensor_category: str | None = None,
                         start: datetime | None = None,
                         end: datetime | None = None,
                         is_current_month: int = 0,
                         not_recovery: int | None = None,
                         limit: int | None = None,
                         page_num: int = 1,
                         page_size: int = 10) -> Resp:
    """
    分页查询警告信息
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
        InFluxDBAlertOrm, 
        SensorOrm.id, 
        SensorOrm.alias, 
        SensorOrm.sensor_category
    ).outerjoin_from(
        InFluxDBAlertOrm, SensorOrm, and_(
            InFluxDBAlertOrm.uuid == SensorOrm.uuid,
            InFluxDBAlertOrm.group_id == SensorOrm.group_id,
            InFluxDBAlertOrm.sensor_sn == SensorOrm.sensor_sn,
        )
    ).where(
        SensorOrm.is_deleted == 0,
    )
    if customer_ids and len(customer_ids) > 0:
        s = s.where(InFluxDBAlertOrm.customer_id.in_(customer_ids))
    if store_ids:
        s = s.where(InFluxDBAlertOrm.store_id.in_(store_ids))
    if type:
        s = s.where(InFluxDBAlertOrm.type == type)
    if status is not None:
        s = s.where(InFluxDBAlertOrm.status == status)
    if uuid is not None:
        s = s.where(InFluxDBAlertOrm.uuid == uuid)
    if group_id is not None:
        s = s.where(InFluxDBAlertOrm.group_id == group_id)
    if start:
        s = s.where(InFluxDBAlertOrm.alert_time >= start)
    if end:
        s = s.where(InFluxDBAlertOrm.alert_time <= end)
    if priority:
        s = s.where(InFluxDBAlertOrm.priority == priority)
    if not_recovery is not None:
        if not_recovery == 1:
            s = s.where(InFluxDBAlertOrm.recovery_time == None)
        else:
            s = s.where(InFluxDBAlertOrm.recovery_time != None)
    if sensor_category:
        s = s.where(SensorOrm.sensor_category == sensor_category)
    if sensor_id:
        s = s.where(SensorOrm.id == sensor_id)
    if is_current_month:
        today = date.today()
        y, m = today.year, today.month
        s = s.where(and_(
            extract("YEAR", InFluxDBAlertOrm.alert_time) == y,
            extract("MONTH", InFluxDBAlertOrm.alert_time) == m
        ))

    total = db_session.scalar(select(func.count()).select_from(s))
    s = s.order_by(InFluxDBAlertOrm.alert_time.desc())
    s = s.offset((page_num - 1) * page_size).limit(
        limit if limit else page_size
    )
    result = db_session.execute(s).fetchall()
    store_ids = set()
    if len(result):
        data = []   
        for x, sensor_id, sensor_alias, sensor_category in result:
            temp = x.to_dict(to_plain=True)
            temp['sensor_id'] = sensor_id
            temp['sensor_alias'] = sensor_alias
            temp['sensor_category'] = sensor_category
            data.append(temp)
            store_ids.add(x.store_id)
        db_session.close()
    else:
        data= result
    resp.data = paging_data(data=data, total=total, page_num=page_num, page_size=page_size)
    resp.data['store_ids'] = list(store_ids)
    return resp


def page_temperature_sensors(keyword: str | None = None,
                             customer_id: int | None = None,
                             store_id: int | None = None,
                             uuid: str | None = None,
                             group_id: int | None = None,
                             page_num: int = 1,
                             page_size: int = 10) -> Resp:
    """
    分页获取温度传感器列表
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
            TemperatureSensorOrm.id,
            TemperatureSensorOrm,
               func.sum(case((and_(InFluxDBAlertOrm.status == 1, InFluxDBAlertOrm.type == SensorEventType.THRESHOLD.value), 1), else_=0)).label("threshold_alert_count"),
               func.sum(case((and_(InFluxDBAlertOrm.status == 1, InFluxDBAlertOrm.type == SensorEventType.OFFLINE.value), 1), else_=0)).label("offline_alert_count"),
               func.count(InFluxDBAlertOrm.id).label("offline_alert_count"),
               ).outerjoin_from(
                   TemperatureSensorOrm, InFluxDBAlertOrm, and_(
                       TemperatureSensorOrm.uuid == InFluxDBAlertOrm.uuid,
                       TemperatureSensorOrm.group_id == InFluxDBAlertOrm.group_id,
                       TemperatureSensorOrm.sensor_sn == InFluxDBAlertOrm.sensor_sn,
                   )
               ).where(
                   TemperatureSensorOrm.is_deleted == 0,
               ).group_by(TemperatureSensorOrm.id, SensorOrm.id)  # 父类的id也必须出现在这里，否则会报错
    if keyword:
        s = s.where(or_(
            TemperatureSensorOrm.alias.ilike(f"%{keyword}%"),
            TemperatureSensorOrm.desc.ilike(f"%{keyword}%"),
            TemperatureSensorOrm.sensor_name.ilike(f"%{keyword}%"),
        ))
    if customer_id: 
        s = s.where(TemperatureSensorOrm.customer_id == customer_id)
    if store_id: 
        s = s.where(TemperatureSensorOrm.store_id == store_id)
    if uuid: 
        s = s.where(TemperatureSensorOrm.uuid == uuid)
    if group_id: 
        s = s.where(TemperatureSensorOrm.group_id == group_id)
    total_num = db_session.scalar(select(func.count()).select_from(s))
    s = s.order_by(TemperatureSensorOrm.create_time.desc()).offset((page_num - 1) * page_size).limit(page_size)
    result = []
    store_ids, uuids, group_ids, sensor_sns = set(), set(), set(), set()
    for _id, x, threshold_alert_count, offline_alert_count, alert_history_count in db_session.execute(s).fetchall():
        temp = x.to_dict(to_plain=True)
        temp['threshold_alert_count'] = threshold_alert_count
        temp['offline_alert_count'] = offline_alert_count
        temp['alert_history_count'] = alert_history_count
        if temp['is_offline'] == 1:
            status = "离线"
        elif temp['is_over_threshold'] == 1:
            status = "超过阈值"
        else:
            status = "正常"
        temp['status'] = status
        result.append(temp)
        store_ids.add(x.store_id)
        uuids.add(x.uuid)
        group_ids.add(x.group_id)
        sensor_sns.add(x.sensor_sn)
    
    data = paging_data(data=result, total=total_num, page_num=page_num, page_size=page_size)
    data['store_ids'] = list(store_ids)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    data['uuids'] = list(uuids)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    data['group_ids'] = list(group_ids)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    data['sensor_sns'] = list(sensor_sns)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    resp.data = data
    return resp


def page_electricity_sensors(keyword: str | None = None,
                             customer_id: int | None = None,
                             store_id: int | None = None,
                             uuid: str | None = None,
                             group_id: int | None = None,
                             ignore_main: int = 0,
                             has_category: int = 0,
                             page_num: int = 1,
                             page_size: int = 10) -> Resp:
    """
    分页获取电量传感器列表
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
            ElectricitySensorOrm.id,
            ElectricitySensorOrm,
               func.sum(case((and_(InFluxDBAlertOrm.status == 1, InFluxDBAlertOrm.type != ElectricitySensorEventType.OFFLINE.value), 1), else_=0)).label("threshold_alert_count"),
               func.sum(case((and_(InFluxDBAlertOrm.status == 1, InFluxDBAlertOrm.type == ElectricitySensorEventType.OFFLINE.value), 1), else_=0)).label("offline_alert_count"),
               func.count(InFluxDBAlertOrm.id).label("offline_alert_count"),
               ElectricitySensorCategoryOrm.name.label("category_name"),
               ).outerjoin_from(
                   ElectricitySensorOrm, InFluxDBAlertOrm, and_(
                       ElectricitySensorOrm.uuid == InFluxDBAlertOrm.uuid,
                       ElectricitySensorOrm.group_id == InFluxDBAlertOrm.group_id,
                       ElectricitySensorOrm.sensor_sn == InFluxDBAlertOrm.sensor_sn,
                   )
               ).outerjoin(
                   ElectricitySensorCategoryOrm, ElectricitySensorOrm.category_id == ElectricitySensorCategoryOrm.id
               ).where(
                   ElectricitySensorOrm.is_deleted == 0,
               ).group_by(ElectricitySensorCategoryOrm.name, ElectricitySensorOrm.id, SensorOrm.id)  # 父类的id也必须出现在这里，否则会报错
    if keyword:
        s = s.where(or_(
            ElectricitySensorOrm.alias.ilike(f"%{keyword}%"),
            ElectricitySensorOrm.desc.ilike(f"%{keyword}%"),
            ElectricitySensorOrm.sensor_name.ilike(f"%{keyword}%"),
        ))
    if customer_id: 
        s = s.where(ElectricitySensorOrm.customer_id == customer_id)
    if store_id: 
        s = s.where(ElectricitySensorOrm.store_id == store_id)
    if uuid: 
        s = s.where(ElectricitySensorOrm.uuid == uuid)
    if group_id: 
        s = s.where(ElectricitySensorOrm.group_id == group_id)
    if ignore_main: 
        s = s.where(ElectricitySensorOrm.is_main == 0)
    if has_category:
        s = s.where(ElectricitySensorOrm.category_id != None)
    total = db_session.scalar(select(func.count()).select_from(s))
    s = s.order_by(ElectricitySensorOrm.create_time.desc()).offset((page_num - 1) * page_size).limit(page_size)
    result = []
    store_ids, uuids, group_ids, sensor_sns = set(), set(), set(), set()
    sensors = db_session.execute(s).fetchall()
    for _id, x, threshold_alert_count, offline_alert_count, alert_history_count, category_name in sensors:
        temp = x.to_dict(to_plain=True)
        temp['threshold_alert_count'] = threshold_alert_count
        temp['offline_alert_count'] = offline_alert_count
        temp['alert_history_count'] = alert_history_count
        temp['category_name'] = category_name
        if temp['is_offline'] == 1:
            status = "离线"
        elif temp['is_over_threshold'] == 1:
            status = "超过阈值"
        else:
            status = "正常"
        temp['status'] = status
        result.append(temp)
        store_ids.add(x.store_id)
        uuids.add(x.uuid)
        group_ids.add(x.group_id)
        sensor_sns.add(x.sensor_sn)
    
    data = paging_data(data=result, total=total, page_num=page_num, page_size=page_size)
    data['store_ids'] = list(store_ids)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    data['uuids'] = list(uuids)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    data['group_ids'] = list(group_ids)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    data['sensor_sns'] = list(sensor_sns)  # 用于给aiops后台查询主机，门店和品牌信息和最新数据
    resp.data = data
    return resp

    
def get_main_electricity_sensors(customer_ids: List[int] | None = None,
                                 store_ids: List[int] | None = None,) -> Resp:
    """
    根据门店id查询他们对应的主电量传感器信息
    :param store_ids: _description_
    :return: _description_
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
        ElectricitySensorOrm.customer_id,
        ElectricitySensorOrm.store_id,
        ElectricitySensorOrm.uuid,
        ElectricitySensorOrm.group_id,
        ElectricitySensorOrm.sensor_sn,
        ).where(
            ElectricitySensorOrm.is_main == 1,
            ElectricitySensorOrm.is_deleted == 0
        )
    if customer_ids:
        s = s.where(ElectricitySensorOrm.customer_id.in_(customer_ids))
    if store_ids:
        s = s.where(ElectricitySensorOrm.store_id.in_(store_ids))
    data = {}
    for customer_id, store_id, uuid, group_id,main_sensor_sn in db_session.execute(s).fetchall():
        container = data.setdefault(customer_id, {})
        container[store_id] = {
            "store_id": store_id,
            "uuid": uuid,
            "group_id": group_id,
            "sensor_sn": main_sensor_sn,
        }
    new_data = []
    for k, v in data.items():
        new_data.append({
            "customer_id": k,
            "stores": list(v.values()),
        })
    db_session.close()
    resp.data = new_data
    return resp


def get_sensor_status_by_id(sensor_id: int) -> Resp:
    """
    根据传感器id查询传感器状态
    :param sensor_id: _description_
    :return: _description_
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
        SensorOrm.status,
    ).where(
        SensorOrm.id == sensor_id, SensorOrm.is_deleted == 0
    )
    status = db_session.scalar(s)
    db_session.close()
    if status is None:
        resp.message = "传感器不存在"
    else:
        resp.data = status
    return resp


def count_sensor_simple(customer_id: int | None = None, store_id: int | None = None) -> Resp:
    """
    统计传感器数量
    """
    db_session, resp = new_db_session(), Resp()
    s = select(
        SensorOrm.sensor_category,
        func.coalesce(func.count(SensorOrm.id), 0),
        func.coalesce(func.sum(SensorOrm.is_offline), 0),
        ).where(
        SensorOrm.is_deleted == 0,
        ).group_by(SensorOrm.sensor_category)
    if customer_id:
        s = s.where(SensorOrm.customer_id == customer_id)
    if store_id:
        s = s.where(SensorOrm.store_id == store_id)
    result = db_session.execute(s)
    data = []
    for sensor_category, total, offline in result:
        online = total - offline
        data.append({
            "sensor_category": sensor_category,
            "total": total,
            "online": online,
            "offline": offline,
            "online_rate": round(online / total, 2) if total else 0,
        })
    resp.data = data
    db_session.close()
    return resp


def count_sensor_category_status(customer_ids: List[int] | None = None, 
                                 store_ids: List[int] | None = None) -> Resp:
    """
    按照传感器分类和状态统计传感器数量
    """
    s = select(
        SensorOrm.sensor_category,
        func.coalesce(func.count(SensorOrm.id)).label("ok"),
        func.coalesce(func.sum(case((SensorOrm.is_offline == 1, 1), else_=0)), 0).label("offline"),
        func.coalesce(func.sum(case((
            and_(
                SensorOrm.is_over_threshold == 1,
                SensorOrm.is_offline == 0
            )
            , 1), else_=0)), 0).label("over_threshold"),
    ).where(
        SensorOrm.is_deleted == 0,
    )
    if customer_ids and len(customer_ids) > 0:
        s = s.where(SensorOrm.customer_id.in_(customer_ids))
    if store_ids:
        s = s.where(SensorOrm.store_id.in_(store_ids))
    s = s.group_by(SensorOrm.sensor_category)
    db_session = new_db_session()
    result = db_session.execute(s).all()
    db_session.close()
    resp = Resp()
    resp.data = {
        x[0]: {
            "total": x[1],
            "ok": x[1] - x[2] - x[3],
            "offline": x[2],
            "over_threshold": x[3],
        } for x in result
    }
    return resp



def count_sensor_alert_quantity(customer_ids: List[int] | None = None, 
                                store_id: int | None = None,
                                ignore_store_ids: List[int] | None = None,
                                days: int = 30) -> Resp:
    """
    统计传感器报警数量
    """
    db_session, resp = new_db_session(), Resp()
    today = datetime.now()
    y, m = today.year, today.month
    s = select(
        InFluxDBAlertOrm.sensor_category,
        func.coalesce(func.count(InFluxDBAlertOrm.id), 0),
        ).where(
            extract("YEAR", InFluxDBAlertOrm.alert_time) == y,
            extract("MONTH", InFluxDBAlertOrm.alert_time) == m,
        )
    if customer_ids and len(customer_ids) > 0:
        s = s.where(InFluxDBAlertOrm.customer_id.in_(customer_ids))
    if store_id:
        s = s.where(InFluxDBAlertOrm.store_id == store_id)
    if ignore_store_ids:
        s = s.where(InFluxDBAlertOrm.store_id.not_in(ignore_store_ids))
    s = s.group_by(InFluxDBAlertOrm.sensor_category)
    result = [{"sensor_category": x[0], "count": x[1]} for x in db_session.execute(s).all()]
    db_session.close()
    resp.data = result
    return resp
    


############### 以下使用 API 而非客户端的方式的查询Demo，注意，比直接使用客户端要慢，只是保留用作参考 #################



def get_iot_temperature_data_by_api(bucket_name: str | None = None, table_name: str | None = None,
                                    uuids: List[str] | None = None, group_ids: List[str] | None = None,
                                    sensor_ids: List[str] | None = None, to_local_time: int = 0,
                                    start: datetime | None = None, max_point: int = 1000,
                                    stop: datetime | None = None) -> Resp:
    """
    获取温度传感器温度数据,这里是通过调用 influxdb 的 API 而非客户端的方式
    此函数目前没有用于任何地方，因为测试表明，API 比直接调用客户端查询要慢。多了接近100毫秒的时间
    :param bucket_name: 数据库
    :param table_name: 表名
    :param uuids: uuid列表
    :param start: 开始时间
    :param stop: 结束时间
    :param max_point: 最多分多少个数据， 会影响返回速度和精度，也会决定返回数据的最大数量
    :return:
    """
    resp = Resp()
    bucket_name = "wn1010" if bucket_name is None else bucket_name
    table_name = "temperature" if table_name is None else table_name
    start = datetime.now() - timedelta(days=1) if start is None else start
    stop = datetime.now() if stop is None else stop
    interval = int((stop - start).total_seconds() / max_point / 60)
    logger.debug(f"设置聚合时间间隔为 {interval} 分钟")
    start = start.strftime("%Y-%m-%dT%H:%M:%SZ")
    stop = stop.strftime("%Y-%m-%dT%H:%M:%SZ")
    query_api = INFLUX_CLI.query_api()
    query = f"""from(bucket:"{bucket_name}")
      |> range(start: {start}, stop: {stop})
      |> filter(fn: (r) => r._measurement == "{table_name}")
      """
    if uuids:
        'r["uuid"] == "30FFD2053054483209770343" or r["uuid"] == "30FFCF053054483213600343"'
        container = [f'r["uuid"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    if group_ids:
        container = [f'r["grp"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    if sensor_ids:
        container = [f'r["id"] == "{x}"' for x in uuids]
        part = " or ".join(container)
        query += f'|> filter(fn: (r) => {part})\n'
    query += f'|> filter(fn: (r) => r["_field"] == "t")\n'  # 只取温度数据
    query += f'|> keep(columns: ["_time", "_value", "uuid", "grp", "id"])\n'  # 只保留时间、温度、uuid、grp、id字段 必须放在 filter之后
    query += f'|> aggregateWindow(every: {interval}m, fn: mean, createEmpty: false)\n'  # 每隔2分钟取一次平均值
    query += f'|> yield(name: "mean")\n'  # 将结果命名为mean
    # 调用 API
    response = None
    url = INFLUXDB_URL + "/api/v2/query"
    params = {"orgID": ORG_ID}
    response = requests.post(url=url, params=params, headers=API_HEADER, data=query)
    response  # 这里返回的是csv 格式，需要转换成json格式
    ...



if __name__ == '__main__':
    count_sensor_simple(67)
    pass

